package de.uni_stuttgart.ipvs.ids.globalstate;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * A process which can transfer & receive money and record & send its state.
 * 
 * @author Team 3/4 Elkhan Dadashov, Ahmed Labib, Dhaval A Shah, Patrick Stortz
 */
public class Process implements Runnable {

	private static final int CONNECT_TIMEOUT = 5000;
	private int id;
	private double balance;
	private int listenPort;
	private ArrayList<Process> neighbors;
	private ArrayList<Collector> collectors;

	private RecordedState recordedState;
	/**
	 * Used to record the money amount in the channels. Key is the process ID of
	 * the sender.
	 */
	private HashMap<Integer, Double> recordedAmountsInChannels;

	/**
	 * Create a process to transfer & receive money and record & send its state.
	 * 
	 * @param id
	 *            The ID of the process
	 * @param balance
	 *            The initial balance of the process
	 * @param listenPort
	 *            The port the socket of this process will listen to
	 */
	public Process(int id, double balance, int listenPort) {
		this.id = id;
		this.balance = balance;
		this.listenPort = listenPort;
		neighbors = new ArrayList<Process>();
		collectors = new ArrayList<Collector>();
		System.out.println("Process " + id + " got created with a balance of "
				+ balance);// DEBUG
	}

	/**
	 * Add a neighbor process to which this process can communicate
	 * 
	 * @param p
	 *            the neighbor process this process can communicate to
	 */
	public void addNeighbour(Process p) {
		if (!neighbors.contains(p))
			neighbors.add(p);
	}

	/**
	 * Register a collector to send a recorded state to
	 * 
	 * @param c
	 *            the collector process to send a recorded state to
	 */
	public void registerCollector(Collector c) {
		if (!collectors.contains(c))
			collectors.add(c);
	}

	/**
	 * Saves the current state of the Process
	 */
	public void saveCurrentState() {
		// behave like we received a marker
		saveCurrentState(null);
	}

	/**
	 * Transfer money to a neighboring account
	 * 
	 * @param amount
	 *            The amount to transfer
	 * @param receiver
	 *            The receiver for the money. Has to be a neighbor, otherwise
	 *            the method just returns.
	 */
	public void transferMoney(double amount, Process receiver) {
		if (!neighbors.contains(receiver))
			return;

		try {
			// create the message
			MoneyMessage transfer = new MoneyMessage(amount, this.getID());
			// send the message
			sendMessage(transfer, receiver.getAddress(), receiver.getPort());
			// message sent, subtract the amount from our balance
			this.balance = this.balance - amount;
			System.out.println("Process " + id + " has a new balance of "
					+ balance + "$ after transfering " + amount
					+ "$ to process " + receiver.getID());// DEBUG
		} catch (IOException e) {
			System.err.println("Error! Process " + id
					+ " cannot transfer money to process " + receiver.getID()
					+ " @ " + receiver.getAddress() + ":" + receiver.getPort()
					+ "!");// DEBUG
			e.printStackTrace();
		}
	}

	@Override
	public void run() {
		try {
			// create our listening socket
			ServerSocket server = new ServerSocket(listenPort);
			ObjectInputStream input = null;
			Object message = null;
			while (true) {
				// wait for incoming connections
				Socket client = server.accept();
				Thread.sleep(100); // simulate delay - do not change

				// get the input stream of the socket connection
				input = new ObjectInputStream(client.getInputStream());

				// read the message from the stream
				message = input.readObject();

				// check if we got a money or marker message
				if (message instanceof MoneyMessage) {
					// part a), process the money message
					processMoneyMessage((MoneyMessage) message);
					// part b), record the channel state
					recordChannelState((MoneyMessage) message);
				} else if (message instanceof MarkerMessage) {
					// part b) + c), start recording of state
					saveCurrentState((MarkerMessage) message);
				} else {
					System.err.println("Error! Process " + id
							+ " received an unknown message!");// DEBUG
				}

				// close the input stream and the socket
				input.close();
				client.close();
			}
		} catch (IOException e) {
			System.err.println("Error! Process " + id
					+ " cannot receive money from a process!");// DEBUG
			e.printStackTrace();
		} catch (InterruptedException e) {
			System.err.println("Error! Process " + id
					+ " got interrupted while waiting for a connection!");// DEBUG
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			System.err.println("Error! Process " + id
					+ " cannot receive money from a process!");// DEBUG
			e.printStackTrace();
		}
	}

	/**
	 * Processes a received MoneyMessage over a socket connection
	 * 
	 * @param message
	 *            the MoneyMessage to be processed
	 */
	private void processMoneyMessage(MoneyMessage message) {
		// add the received message to our balance
		this.balance = this.balance + message.amount;
		System.out.println("Process " + id + " has a new balance of " + balance
				+ "$ after receiving " + message.amount + "$ from process "
				+ message.getSender());// DEBUG
	}

	/**
	 * Sends a message to another Process
	 * 
	 * @param message
	 *            the message to be sent
	 * @param process
	 *            the Process the message has to be sent to
	 * @throws IOException
	 *             thrown if an error occures while sending the message
	 */
	private void sendMessage(Object message, String address, int port)
			throws IOException {
		// create our socket and connect to the destination socket
		Socket clientSocket = new Socket();
		clientSocket.connect(new InetSocketAddress(address, port),
				CONNECT_TIMEOUT);

		// check if the socket is properly connected
		if (clientSocket.isConnected() && clientSocket.isClosed() == false) {
			// get the output stream to the receiving socket
			ObjectOutputStream output = new ObjectOutputStream(
					clientSocket.getOutputStream());
			// send the message
			output.writeObject(message);
			// flush and close the stream
			output.flush();
			output.close();
		}

		// close the socket
		clientSocket.close();

	}

	/**
	 * Saves the current state after receiving a MarkerMessage
	 * 
	 * @param marker
	 *            the MarkerMessage received or null if it is the start of the
	 *            algorithm
	 */
	private void saveCurrentState(MarkerMessage marker) {

		if (recordedState == null) {
			// first time we record our state

			// receiving rule:
			// 1. record our state
			// 2. record the state of this incoming channel as empty set
			// 3. turn on recording of messages arriving over other incoming
			// channels

			// record our state
			recordedState = new RecordedState(this.id, this.balance);

			// record the state of this incoming channel as empty set
			if (marker != null) {
				recordedState.setAmountInSingleChannel(marker.getSender(), 0.0);
			}

			// turn on recording of messages arriving over other incoming
			// channels
			recordedAmountsInChannels = new HashMap<Integer, Double>();
			for (Process neighbor : neighbors) {
				if (marker != null) {
					if (marker.getSender() != neighbor.getID()) {
						recordedAmountsInChannels.put(neighbor.getID(), 0.0);
					}
				} else {
					recordedAmountsInChannels.put(neighbor.getID(), 0.0);
				}

			}

			// sending rule:
			// after recording our state send a marker message message over our
			// outgoing channels

			// create a marker message and send it over all outgoing channels
			MarkerMessage markerMessage = new MarkerMessage(this.getID());
			for (Process neighbor : neighbors) {
				try {
					sendMessage(markerMessage, neighbor.getAddress(),
							neighbor.getPort());
				} catch (IOException e) {
					System.err.println("Error! Process " + id
							+ " cannot send marker to process "
							+ neighbor.getID() + " @ " + neighbor.getAddress()
							+ ":" + neighbor.getPort() + "!");// DEBUG
				}
			}

		} else {
			// record the state of this incoming channel as set of messages we
			// have received over it since we saved our state and turn off
			// recording

			// check if this is not the start of the algorithm
			if (marker != null) {
				// check if we ever started recording the state of this channel
				if (recordedAmountsInChannels.containsKey(marker.getSender())) {
					// set the state of this channel as the sum of amounts we
					// have received over it since we saved our state
					recordedState.setAmountInSingleChannel(marker.getSender(),
							recordedAmountsInChannels.get(marker.getSender()));

					// turn off the recording of this channel
					recordedAmountsInChannels.remove(marker.getSender());
				}
			}

			// check if we finished recording for all of our channels
			if (recordedAmountsInChannels.isEmpty()) {
				// we finished, algorithm has to terminate, 
				// tell our collectors that we finished recording our state
				for (Collector collector : collectors) {
					try {
						sendMessage(this.recordedState, collector.getAddress(),
								collector.getPort());
						System.out.println("Process " + id + " sent state ("
								+ recordedState + ") to collector");// DEBUG
					} catch (IOException e) {
						System.err.println("Error! Process " + id
								+ " cannot send state to collector @ "
								+ collector.getAddress() + ":"
								+ collector.getPort() + "!");// DEBUG
					}
				}
			}
		}

	}

	private void recordChannelState(MoneyMessage message) {
		// check if we need to record incoming messages of this channel
		if (recordedAmountsInChannels.containsKey(message.getSender())) {
			// get the current amount in this channel
			double oldChannelAmount = recordedAmountsInChannels
					.get(message.getSender());
			// update the amount in this channel
			recordedAmountsInChannels.put(message.getSender(), oldChannelAmount
					+ message.getAmount());
		}
	}

	/**
	 * Get the host address of this process
	 * 
	 * @return the host address of this process
	 */
	public String getAddress() {
		try {
			return java.net.InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			e.printStackTrace();
			return "localhost";
		}
	}

	/**
	 * Get the listening port of this process
	 * 
	 * @return the listening port of this process
	 */
	public int getPort() {
		return listenPort;
	}

	/**
	 * Get the ID of this process
	 * 
	 * @return the ID of this process
	 */
	public int getID() {
		return id;
	}

	/**
	 * Get the current balance of this process
	 * 
	 * @return the current balance of this process
	 */
	public double getBalance() {
		return balance;
	}
}
